package drds.server.route.sequence;

import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import drds.server.Server;
import drds.server.config.Config;
import drds.server.config.util.ConfigException;
import drds.server.datasource.GroupDataSourceNode;
import drds.server.datasource.IConnectionHolder;
import drds.server.handler.Handler;
import drds.server.net.packet.ErrorPacket;
import drds.server.net.packet.RowDataPacket;
import drds.server.route.RouteNode;
import drds.server.route.util.PropertiesUtil;
import drds.server.server.parser.ServerParse;

public class SequenceHandlerImpl implements SequenceHandler {

	protected static final Logger LOGGER = LoggerFactory.getLogger(SequenceHandlerImpl.class);

	private static final String DB_SEQUENCE_PROPERTIES = "db_sequence.properties";
	protected static final String ERR_SEQ_RESULT = "-999999999,null";
	protected static Map<String, String> latestErrors = new ConcurrentHashMap<String, String>();
	private final DBSequnceHandler dataBaseSequnceHandler = new DBSequnceHandler();

	private static class IncrSequenceMySQLHandlerHolder {
		private static final SequenceHandlerImpl instance = new SequenceHandlerImpl();
	}

	public static SequenceHandlerImpl getInstance() {
		return IncrSequenceMySQLHandlerHolder.instance;
	}

	public SequenceHandlerImpl() {

		load();
	}

	public void load() {
		// load sequnce properties
		Properties properties = PropertiesUtil.loadProperties(DB_SEQUENCE_PROPERTIES);
		removeDesertedSequenceVals(properties);
		putNewSequenceVals(properties);
	}

	private void removeDesertedSequenceVals(Properties properties) {
		Iterator<Map.Entry<String, SequenceValue>> iterator = sequenceValueMap.entrySet().iterator();
		while (iterator.hasNext()) {
			Map.Entry<String, SequenceValue> entry = iterator.next();
			if (!properties.containsKey(entry.getKey())) {
				iterator.remove();
			}
		}
	}

	private void putNewSequenceVals(Properties properties) {
		for (Map.Entry<Object, Object> entrySet : properties.entrySet()) {
			String sequenceName = (String) entrySet.getKey();
			String dataNode = (String) entrySet.getValue();
			if (!sequenceValueMap.containsKey(sequenceName)) {
				sequenceValueMap.put(sequenceName, new SequenceValue(sequenceName, dataNode));
			} else {
				sequenceValueMap.get(sequenceName).dataNode = dataNode;
			}
		}
	}

	/**
	 * save sequnce -> curval
	 */
	private ConcurrentHashMap<String, SequenceValue> sequenceValueMap = new ConcurrentHashMap<String, SequenceValue>();

	@Override
	public long nextId(String sequenceName) {
		SequenceValue sequenceValue = sequenceValueMap.get(sequenceName);
		if (sequenceValue == null) {
			throw new ConfigException("can't find definition for sequence :" + sequenceName);
		}
		if (!sequenceValue.isSuccessFetched()) {
			return getSequenceValueFromDB(sequenceValue);
		} else {
			return getNextSequenceValue(sequenceValue);
		}

	}

	private Long getNextSequenceValue(SequenceValue sequenceValue) {
		Long nextValue = sequenceValue.nextValue();
		if (sequenceValue.isNextValidValid(nextValue)) {
			return nextValue;
		} else {
			sequenceValue.fetching.compareAndSet(true, false);
			return getSequenceValueFromDB(sequenceValue);
		}
	}

	private long getSequenceValueFromDB(SequenceValue sequenceValue) {
		if (sequenceValue.fetching.compareAndSet(false, true)) {
			sequenceValue.dbReturnValue = null;
			sequenceValue.dbFinished = false;
			sequenceValue.newValueSetted.set(false);
			dataBaseSequnceHandler.execute(sequenceValue);
		}
		Long[] values = sequenceValue.waitFinish();
		if (values == null) {

			throw new RuntimeException("can't fetch sequnce in db,sequnce :" + sequenceValue.sequenceName + " detail:" + dataBaseSequnceHandler.getLastestError(sequenceValue.sequenceName));
		} else {
			if (sequenceValue.newValueSetted.compareAndSet(false, true)) {
				sequenceValue.setCurrentValue(values[0]);
				sequenceValue.maxSegmentValue = values[1];
				return values[0];
			} else {
				return sequenceValue.nextValue();
			}

		}

	}
}

class DBSequnceHandler implements Handler {
	private static final Logger LOGGER = LoggerFactory.getLogger(DBSequnceHandler.class);

	public void execute(SequenceValue sequenceValue) {
		Config config = Server.getInstance().getConfig();
		GroupDataSourceNode groupDataSourceNode = config.getGroupDataSourceNodeMap().get(sequenceValue.dataNode);
		try {
			// 修正获取seq的逻辑，在读写分离的情况下只能走写节点。修改Select模式为Update模式。
			groupDataSourceNode.getConnection(groupDataSourceNode.getSchemaName(), true, new RouteNode(sequenceValue.dataNode, ServerParse.UPDATE, sequenceValue.sql), this, sequenceValue);
		} catch (Exception e) {
			LOGGER.error("get connection err " + e);
		}

	}

	public String getLastestError(String seqName) {
		return SequenceHandlerImpl.latestErrors.get(seqName);
	}

	@Override
	public void connectionAcquired(IConnectionHolder connectionHolder) {

		connectionHolder.setHandler(this);
		try {
			connectionHolder.query(((SequenceValue) connectionHolder.getAttachment()).sql);
		} catch (Exception e) {
			executeException(connectionHolder, e);
		}
	}

	@Override
	public void connectionError(IConnectionHolder connectionHolder, Throwable e) {
		((SequenceValue) connectionHolder.getAttachment()).dbFinished = true;
		LOGGER.error("connectionError " + e);

	}

	@Override
	public void errorResponse(IConnectionHolder connectionHolder, byte[] bytes) {
		SequenceValue sequenceValue = ((SequenceValue) connectionHolder.getAttachment());
		sequenceValue.dbFinished = true;

		ErrorPacket errorPacket = new ErrorPacket();
		errorPacket.read(bytes);
		String errMsg = new String(errorPacket.errorMessage);
		LOGGER.error("errorResponse " + errorPacket.errorNo + " " + errMsg);
		SequenceHandlerImpl.latestErrors.put(sequenceValue.sequenceName, errMsg);
		connectionHolder.release();

	}

	@Override
	public void okResponse(IConnectionHolder connectionHolder, byte[] okBytes) {
		boolean executeResponse = connectionHolder.syncAndExcute();
		if (executeResponse) {
			((SequenceValue) connectionHolder.getAttachment()).dbFinished = true;
			connectionHolder.release();
		}

	}

	@Override
	public void rowResponse(IConnectionHolder connectionHolder, byte[] rowBytes) {
		RowDataPacket rowDataPkg = new RowDataPacket(1);
		rowDataPkg.read(rowBytes);
		byte[] columnValueBytes = rowDataPkg.fieldBytesList.get(0);
		String columnValue = new String(columnValueBytes);
		SequenceValue sequenceValue = (SequenceValue) connectionHolder.getAttachment();
		if (SequenceHandlerImpl.ERR_SEQ_RESULT.equals(columnValue)) {
			sequenceValue.dbReturnValue = SequenceHandlerImpl.ERR_SEQ_RESULT;
			LOGGER.error(" sequnce sql returned err value ,sequence:" + sequenceValue.sequenceName + " " + columnValue + " sql:" + sequenceValue.sql);
		} else {
			sequenceValue.dbReturnValue = columnValue;
		}
	}

	@Override
	public void rowEofResponse(IConnectionHolder connectionHolder, byte[] eofBytes) {
		((SequenceValue) connectionHolder.getAttachment()).dbFinished = true;
		connectionHolder.release();
	}

	private void executeException(IConnectionHolder connectionHolder, Throwable e) {
		SequenceValue sequenceValue = ((SequenceValue) connectionHolder.getAttachment());
		sequenceValue.dbFinished = true;
		String errMgs = e.toString();
		SequenceHandlerImpl.latestErrors.put(sequenceValue.sequenceName, errMgs);
		LOGGER.error("executeException   " + errMgs);
		connectionHolder.close("exception:" + errMgs);

	}

	@Override
	public void writeQueueAvailable() {

	}

	@Override
	public void connectionClose(IConnectionHolder conn, String reason) {

		LOGGER.error("connection closed " + conn + " reason:" + reason);
	}

	@Override
	public void fieldEofResponse(IConnectionHolder conn, byte[] header, List<byte[]> fields, byte[] eof) {

	}

}

class SequenceValue {
	public AtomicBoolean newValueSetted = new AtomicBoolean(false);
	public AtomicLong currentValue = new AtomicLong(0);
	public volatile String dbReturnValue = null;
	public volatile boolean dbFinished;
	public AtomicBoolean fetching = new AtomicBoolean(false);
	public volatile long maxSegmentValue;
	public volatile boolean successFetched;
	public volatile String dataNode;
	public final String sequenceName;
	public final String sql;

	public SequenceValue(String sequenceName, String dataNode) {
		this.sequenceName = sequenceName;
		this.dataNode = dataNode;
		sql = "SELECT next_sequence_value('" + sequenceName + "')";
	}

	public boolean isNextValidValid(Long nexVal) {
		if (nexVal < this.maxSegmentValue) {
			return true;
		} else {
			return false;
		}
	}

	// FetchMySQLSequnceHandler seqHandler;

	public void setCurrentValue(long newValue) {
		currentValue.set(newValue);
		successFetched = true;
	}

	public Long[] waitFinish() {
		long start = System.currentTimeMillis();
		long end = start + 10 * 1000;
		while (System.currentTimeMillis() < end) {
			if (dbReturnValue == SequenceHandlerImpl.ERR_SEQ_RESULT) {
				throw new java.lang.RuntimeException("sequnce not found in db table ");
			} else if (dbReturnValue != null) {
				String[] items = dbReturnValue.split(",");
				Long currentValue = Long.parseLong(items[0]);
				int span = Integer.parseInt(items[1]);
				return new Long[] { currentValue, currentValue + span };
			} else {
				try {
					Thread.sleep(100);
				} catch (InterruptedException e) {
					SequenceHandlerImpl.LOGGER.error("wait db fetch sequnce err " + e);
				}
			}
		}
		return null;
	}

	public boolean isSuccessFetched() {
		return successFetched;
	}

	public long nextValue() {
		if (successFetched == false) {
			throw new java.lang.RuntimeException("sequnce fetched failed  from db ");
		}
		return currentValue.incrementAndGet();
	}
}
